package org.xserver.component.handler;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.lang.reflect.InvocationTargetException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import javax.annotation.Resource;

import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.handler.codec.http.HttpChunk;
import org.jboss.netty.handler.codec.http.HttpHeaders;
import org.jboss.netty.handler.codec.http.HttpRequest;
import org.jboss.netty.handler.codec.http.HttpResponseStatus;
import org.jboss.netty.handler.codec.http.HttpVersion;
import org.jboss.netty.handler.codec.http.multipart.HttpPostRequestDecoder;
import org.jboss.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import org.jboss.netty.handler.ssl.NotSslRecordException;
import org.jboss.netty.handler.ssl.SslHandler;
import org.jboss.netty.handler.timeout.ReadTimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import org.xserver.common.bean.BaseResponse;
import org.xserver.common.util.CharsetUtil;
import org.xserver.common.util.MessageUtils;
import org.xserver.common.util.StringUtil;
import org.xserver.component.config.XServerHttpConfig;
import org.xserver.component.core.XServerHttpContextAttachment;
import org.xserver.component.core.XServerHttpRequest;
import org.xserver.component.core.XServerHttpResponse;
import org.xserver.component.core.dispatch.RequestDispatch;
import org.xserver.component.core.dispatch.RequestDispatchFilter;
import org.xserver.component.core.dispatch.RequestDispatchManager;
import org.xserver.component.core.dispatch.http.WebSocketRequestDispatch;
import org.xserver.component.core.websocket.WebSocketCacheManager;
import org.xserver.component.core.websocket.WebSocketClient;
import org.xserver.component.exception.AbstractBussinessException;
import org.xserver.component.exception.AbstractServerError;
import org.xserver.component.extension.filter.FilterType;
import org.xserver.component.extension.filter.parameter.XServerParametersProcessFilter;
import org.xserver.component.handler.ext.WriteHandler;
import org.xserver.component.handler.ext.WriteHandlerFactory;
import org.xserver.component.handler.ext.websocket.WebsocketWriteHandler;
import org.xserver.component.json.JsonManager;
import org.xserver.component.websocket.XServerWebSocketServerHandshaker13;

import com.fasterxml.jackson.core.JsonProcessingException;

/**
 * Interface dispatch
 * <p>
 * The request URL rule is
 * "http(s)://svr.xserver.com:port/className/methodName?param1=value1&amp;param2=value2"
 * .
 * </p>
 * Frame uses reflection to invoke interfaces method, all default wrapped
 * interfaces have two parameter, one is XServerHttpRequest, another is
 * XServerHttpResponse, this style like Servlet. The business logic use
 * <code>XServerHttpRequest</code>'s <code>getParameter</code>() to grain
 * parameter(s).
 * 
 * the workflow as follow:
 * 
 * <pre>
 *                         XServer
 *               +------------------------+
 *               |        Encoder         |
 *            |  |     Business Logic     | /|\
 * downstream |  |     RequestDispath     |  | upstream
 *           \|/ |        Decoder         |  |
 *               +------------------------+
 *                          Client
 * 
 * </pre>
 * 
 * @author idoldefine
 * @since 2013/01/10
 */

@Component
public class RequestDispatchHandler extends SimpleChannelUpstreamHandler {
	public static final Logger logger = LoggerFactory
			.getLogger(RequestDispatchHandler.class);

	@Resource
	private RequestDispatchFilter requestDispatchFilter;

	@Resource
	private RequestDispatchManager requestDispatchManager;
	@Resource
	private WriteHandlerFactory writeHandlerFactory;
	@Resource
	private XServerHttpConfig httpConfig;
	@Resource
	private XServerParametersProcessFilter xServerParametersProcessFilter;

	private Map<Integer, HttpPostRequestDecoder> context = new ConcurrentHashMap<Integer, HttpPostRequestDecoder>(
			1024);
	@Resource
	private WebSocketCacheManager webSocketCacheManager;
	/**
	 * One request just enter this method only one(exclude HttpChunk)
	 */
	@Override
	public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
			throws Exception {
		Object message = e.getMessage();
		if (message instanceof XServerHttpRequest) {
			XServerHttpRequest request = (XServerHttpRequest) message;
			
			request.setRemoteAddress(e.getRemoteAddress());
			request.setLocalAddress(e.getChannel().getLocalAddress());

			request.setClientAddress(getClientAddress(e, request));

			String format = "Request from {}, clientIP:{}, {} {} requestURI:{}";
			Object[] data = new Object[] { e.getRemoteAddress(),
					request.getClientAddress(), request.getMethod().getName(),
					request.getProtocolVersion(), request.getUri() };
			if (request.isPostMethod()) {
				format = format + ", body:{}";
				data = new Object[] { e.getRemoteAddress(),
						request.getClientAddress(),
						request.getMethod().getName(),
						request.getProtocolVersion(), request.getUri(),
						request.getContent().toString(CharsetUtil.UTF_8) };
			}

			logger.info(format, data);
			
			ctx.setAttachment(request);

			XServerHttpContextAttachment attachment = new XServerHttpContextAttachment(
					request, ctx, httpConfig);
			/**
			 * because the webSocket handshake base on HTTP, so we add the
			 * logical action to HTTP handler but send her to next pipeline
			 */
			if (isWebSocketHandshake(request)) {
				webSocketHandshake(ctx, e);
				attachment.setWebsocket(true);
				return;
			}
			/*
			 * if message is chunked, the server should accept all message body
			 * before into business logic
			 */
			if (request.isChunked()) {
				context.put(e.getChannel().getId(), request.getPostDecoder());
				return;
			}
		    dispatch(attachment);
		}

		/*
		 * if the message is http chunk, we should append rest to the context
		 * message
		 */
		if (message instanceof HttpChunk) {
			HttpChunk httpChunk = (HttpChunk) message;

			XServerHttpRequest request = (XServerHttpRequest) ctx
					.getAttachment();
			HttpPostRequestDecoder decoder = context
					.get(e.getChannel().getId());
			decoder.offer(httpChunk);

			if (httpChunk.isLast()) {
				try {
					dispatch(new XServerHttpContextAttachment(request, ctx,
							httpConfig));
				} finally {
					context.remove(e.getChannel().getId());
				}
			} else {
				return;
			}
		}

		/*
		 * next upstream is WebSocket, if not WebSocket, it will upstream to
		 * next handler
		 */
		ctx.sendUpstream(e);
	}

	private String getClientAddress(MessageEvent e, XServerHttpRequest request) {
		String clientAddress = StringUtils.isEmpty(request.headers().get(
				"X-Forwarded-For")) ? e.getRemoteAddress().toString().split("/")[1] : 
				request.headers().get("X-Forwarded-For");
		return clientAddress;
	}

	/**
	 * before filters -&gt business logic -&gt return filters
	 * 
	 * @param attachment
	 * @throws Exception
	 */
	private void dispatch(XServerHttpContextAttachment attachment)
			throws Exception {

		RequestDispatch requestDispatch = requestDispatchManager
				.getInstance(attachment);
		if(requestDispatch instanceof WebSocketRequestDispatch){
			return;
		}
		WriteHandler writeHandler = writeHandlerFactory.getInstance(attachment);
		if(writeHandler instanceof WebsocketWriteHandler){
			return;
		}
		requestDispatchFilter.filterDispatch(attachment, FilterType.BEFORE);
		/**
		 * if filter has already wrote content to remote client and wouldn't go
		 * into the business logic
		 */
		if (!attachment.hasWrote()) {
			long startTime = System.currentTimeMillis();
			//Object result = 
			requestDispatch.requestDispatch(attachment,writeHandler);
			//writeHandler.writeResponse(attachment, result);
			long offset = System.currentTimeMillis() - startTime;
			logger.info("process uri[{}] uses {} milliseconds", attachment
					.getRequest().getUri(), offset);
		}
		//requestDispatchFilter.filterDispatch(attachment, FilterType.RETURN);
	}

	@Override
	public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e)
			throws Exception {
		e.getChannel().close();
	}

	@Override
	public void channelDisconnected(ChannelHandlerContext ctx,
			ChannelStateEvent e) throws Exception {
		e.getChannel().close();
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
			throws Exception {
		super.exceptionCaught(ctx, e);

		Throwable eCause = e.getCause();

		logger.error("RequestDispatchHandler occur exception", e.getCause());
		if (eCause instanceof ReadTimeoutException) {
			XServerHttpResponse response = new XServerHttpResponse(
					HttpVersion.HTTP_1_1, HttpResponseStatus.FORBIDDEN);
			response.setContentString("Readtime out");
			if (e.getChannel().isOpen()) {
				if (e.getChannel().isWritable()) {
					e.getChannel().write(response)
							.addListener(ChannelFutureListener.CLOSE);
				} else {
					e.getChannel().close();
				}
			}
		} else if (eCause instanceof NotSslRecordException) {
			NotSslRecordException exception = (NotSslRecordException) e
					.getCause();
			XServerHttpResponse response = new XServerHttpResponse(
					HttpVersion.HTTP_1_1, HttpResponseStatus.NOT_ACCEPTABLE);
			response.setContentString(exception.getMessage());
			ctx.getPipeline().remove(SslHandler.class.getSimpleName());
			if (e.getChannel().isOpen()) {
				if (e.getChannel().isWritable()) {
					e.getChannel().write(response)
							.addListener(ChannelFutureListener.CLOSE);
				} else {
					e.getChannel().close();
				}
			}
		} else if (eCause instanceof InvocationTargetException) {
			if (((InvocationTargetException) eCause).getTargetException() instanceof AbstractServerError) {
				AbstractServerError abstractServerError = (AbstractServerError) ((InvocationTargetException) eCause)
						.getTargetException();

				setErrorMessage(abstractServerError.getCode(),
						abstractServerError.getMessage(), e,
						abstractServerError);

			} else if (((InvocationTargetException) eCause)
					.getTargetException() instanceof AbstractBussinessException) {
				AbstractBussinessException bException = (AbstractBussinessException) ((InvocationTargetException) eCause)
						.getTargetException();

				setErrorMessage(bException.getCode(), bException.getMessage(),
						e, bException);
			}else{
				setErrorMessage("500",
						"occur error", e,
						"something null");
			}

		} else if (eCause instanceof AbstractServerError) {

			AbstractServerError abstractServerError = (AbstractServerError) eCause;
			setErrorMessage(abstractServerError.getCode(),
					abstractServerError.getMessage(), e, abstractServerError);
			// BaseResponse<AbstractServerError> baseResponse = new
			// BaseResponse<AbstractServerError>();
			// baseResponse.setErrcode(abstractServerError.getCode());
			// baseResponse.setDescription(abstractServerError.getMessage());
			// baseResponse.setResult(abstractServerError);
			// XServerHttpResponse response = new XServerHttpResponse(
			// HttpVersion.HTTP_1_1, HttpResponseStatus.NOT_ACCEPTABLE);
			// response.setContentString(JsonManager.json(baseResponse));
			// if (e.getChannel().isOpen()) {
			// if (e.getChannel().isWritable()) {
			// e.getChannel().write(response)
			// .addListener(ChannelFutureListener.CLOSE);
			// } else {
			// e.getChannel().close();
			// }
			// }
		} else {
			if (e.getChannel() != null) {
				XServerHttpResponse response = new XServerHttpResponse(
						HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST);
				if (!StringUtil.isEmpty(eCause.getMessage())) {
					response.setContentString(eCause.getMessage());
					if (e.getChannel().isOpen()) {
						if (e.getChannel().isWritable()) {
							e.getChannel().write(response)
									.addListener(ChannelFutureListener.CLOSE);
						} else {
							e.getChannel().close();
						}
					}
				} else {
					StringWriter sw = null;
					PrintWriter ps = null;
					try {
						sw = new StringWriter();
						ps = new PrintWriter(sw);
						eCause.printStackTrace(ps);
						response.setContentString(sw.toString());
						if (e.getChannel().isOpen()) {
							if (e.getChannel().isWritable()) {
								e.getChannel()
										.write(response)
										.addListener(
												ChannelFutureListener.CLOSE);
							} else {
								e.getChannel().close();
							}
						}
					} finally {
						sw.close();
						ps.close();
					}
				}
			}
		}
		// TODO throwableHandler.handleThrowable(attachment, e);
	}

	private void setErrorMessage(String code, String message, ExceptionEvent e,
			Object t) throws JsonProcessingException {
		BaseResponse<Object> baseResponse = new BaseResponse<Object>();
		baseResponse.setErrcode(code);
		baseResponse.setDescription(message);
		baseResponse.setResult(t);
		XServerHttpResponse response = new XServerHttpResponse(
				HttpVersion.HTTP_1_1, HttpResponseStatus.NOT_ACCEPTABLE);
		response.setContentString(JsonManager.json(baseResponse));
		if (e.getChannel().isOpen()) {
			if (e.getChannel().isWritable()) {
				e.getChannel().write(response)
						.addListener(ChannelFutureListener.CLOSE);
			} else {
				e.getChannel().close();
			}
		}
	}

	private WebSocketServerHandshaker webSocketHandshake(
			ChannelHandlerContext ctx, MessageEvent e) {
		XServerHttpRequest request = (XServerHttpRequest) e.getMessage();
		XServerWebSocketServerHandshaker13 handshaker = new XServerWebSocketServerHandshaker13(
				request.getUri(), null, false);
		handshaker.handshake(ctx.getChannel(), request).addListener(
				WebSocketServerHandshaker.HANDSHAKE_LISTENER);
		WebSocketClient webSocketClient = new WebSocketClient();
		webSocketClient.setUri(MessageUtils.getHttpGetUri(request.getUri()));
		// 设置请求参数
		webSocketClient.setReqParam(MessageUtils.getHttpGetParams(request
				.getUri()));
		webSocketClient.setChannelHandlerContext(ctx);
		webSocketClient.setHandshaker(handshaker);
		// 注册连接管理器
		webSocketCacheManager.putWebSocketClient(ctx.getChannel().getId()
				.toString(), webSocketClient);
		return handshaker;
	}

	/**
	 * The current request whether is WebSocket handshake, if WebSocket
	 * handshake return true, else return false.
	 * 
	 * At present, the check WebSocket handshake is necessary but not sufficient
	 * 
	 * @param request
	 *            current request
	 * @return whether WebSocket handshake
	 */
	private boolean isWebSocketHandshake(HttpRequest request) {
		String upgrade = request.headers().get(HttpHeaders.Names.UPGRADE);

		if (WebSocketHandler.WEBSOCKET.equalsIgnoreCase(upgrade)) {
			return true;
		}

		return false;
	}
}
